import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object zy1 {
  def main(args: Array[String]): Unit = {
    //使用IDEA创建Maven工程，添加Spark相关依赖，截图完整；
    //	2、创建SparkCore上下文对象，设置日志级别为ERROR；
    //	3、读取车辆行驶监控日志文件数据，封装集合RDD；
    //	4、计算数据量和打印前10条样本数据
    val sparkConf = new SparkConf().setMaster("local").setAppName("app")
    val sc:SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("ERROR")

    val lineText:RDD[String]=sc.textFile("data/vehicle-log.txt")
    val a:Array[String] =lineText.take(10)
   // for (elem <-a){
    //  println(elem)
    //}
    print("------------------------------------------------------------------")
    //定义样例类VehicleData，封装车辆数据，包含上述所有字段



////    6、使用RDD算子，解析每行数据，封装样例类对象
val mapRDD: RDD[VehicleData] = lineText.filter(a => {
  a.split(",").length == 5
}).map(a => {
  var arr = a.split(",")
  VehicleData(arr(0), arr(1), arr(2), arr(3), arr(4))
})
    //mapRDD.foreach(println)
    //println("------------7----------------")
//7、使用RDD算子，过滤获取行驶速度大于70以上所有数据，打印控制台；（4分）
    var rddFilter:RDD[VehicleData]=mapRDD.filter(a=>a.speed.toInt>70)
    rddFilter.foreach(println)
//8、 将上述结果数据，保存HDFS文件系统，并使用命令查看文件数据；（5分）
    //rddFilter.saveAsTextFile("hdfs://cdh01:8020/tmp/data")

    // 9、使用RDD算子，按照车牌号码中各个省份分组，统计车辆数据的数据量，打印控制台
    var rddGroupby:RDD[(String,Iterable[VehicleData])]=
      rddFilter.groupBy(a=>a.cphm.split(" ")(0))

    var rddGroupbyValues:RDD[(String,Int)]=rddGroupby.map(a=>{
      (a._1,a._2.size)
    })
    //rddGroupbyValues.foreach(println)

    ////    10、对上述结果数据，按照数据量降序排序，打印控制台；（5分）
    var rddGroupbyValuesSortBy:RDD[(String,Int)]=rddGroupbyValues.sortBy(a=>
      {a._2},false
    )
    //rddGroupbyValuesSortBy.foreach(println)
    ////    11、使用RDD算子，按照区域分组，统计车辆数据的数据量，获取Top5区域，打印控制台；（5分）
    var rddGroupbyValuesSortByTop5:Array[(String,Int)]=rddGroupbyValuesSortBy.take(5)
    rddGroupbyValuesSortByTop5.foreach(println)

    sc.stop()
  }
  case class VehicleData (var cphm:String,var dt:String, var speed:String,var area_number:String,var road_number:String)
}
